package com.we.risk.phoneRecord.tools;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;
import com.we.flink.utils.WeKafkaPropertyReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.*;

public class CaptureUserKeyPropTest {
    public static final String RELEASEPROP =
            "risk/phonerecord/stage2/adm_user_label_kfk_prod.properties";
    public static final String PROPSETS = "risk/phonerecord/phonerecord.properties";
    public static Logger LOG = LoggerFactory.getLogger(CaptureUserKeyPropTest.class);

    public static final String DATEFORMAT = "yyyy-MM-dd";
    public static final String INPUTKEYBY = "user_key";
    public static final String TIMESTAMP = "timestamp";
    public static final String CAPTURETIME = "capture_time";
    public static final String OUTKEYBY = "user_key";
    public static final String LABELOUTIME = "label_output_time";
    public static final String OUTJOBID = "job_id";
    public static final String JOBID = "jobID";
    public static final String PHONECALLLIST = "phoneCallList";

    public static void main(String[] args) throws IOException {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        try {
            WeKafkaPropertyReader paramReader = WeKafkaPropertyReader.init(RELEASEPROP);
            /** RocksDB */
            env.setStateBackend(new RocksDBStateBackend(paramReader.getRocksDBBackendUrl()));
            /** checkpoint configure */
            CheckpointConfig ckConf = env.getCheckpointConfig();
            ckConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            ckConf.setCheckpointInterval(10 * 60 * 1000); // ms
            ckConf.setCheckpointTimeout(60 * 60 * 1000);
            ckConf.setMaxConcurrentCheckpoints(1);
            ckConf.setMinPauseBetweenCheckpoints(500);
            ckConf.enableExternalizedCheckpoints(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

            ckConf.enableUnalignedCheckpoints();

            /** Consumer Kafka */
            String srcKafkaTopic = paramReader.getTmpKfkTopic();
            String srcKafkaBootStrapServer = paramReader.getTmpKfkBootStrapServer();
            String srckfkGrupId = paramReader.getTmpKfkGroupId();

            Properties consumProp = new Properties();
            consumProp.setProperty("group.id", srckfkGrupId);
            consumProp.setProperty(
                    "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            consumProp.setProperty(
                    "value.deserializer",
                    "org.apache.kafka.common.serialization.StringDeserializer");

            KafkaSource<String> kfkSource =
                    KafkaSource.<String>builder()
                            .setBootstrapServers(srcKafkaBootStrapServer)
                            .setTopics(srcKafkaTopic)
                            .setStartingOffsets(OffsetsInitializer.earliest())
                            .setProperties(consumProp)
                            .setValueOnlyDeserializer(new SimpleStringSchema())
                            .build();

            SingleOutputStreamOperator<String> input =
                    env.fromSource(kfkSource, WatermarkStrategy.noWatermarks(), "PhoneRecord")
                            .uid("phone-record-stage2-source");

            SingleOutputStreamOperator<String> out =
                    input.flatMap(
                            new RichFlatMapFunction<String, String>() {
                                @Override
                                public void flatMap(String value, Collector<String> out)
                                        throws Exception {
                                    try {
                                        if (!value.contains(JOBID)) {
                                            JSONObject inputJson = JSONObject.parseObject(value);
                                            String user_key = inputJson.getString(INPUTKEYBY);
                                            if (user_key.equals(
                                                    "57702f7cfd736404253c92118cf82b9b")) {
                                                LOG.error("value: " + value);
                                                out.collect(value);
                                            }
                                        }
                                    } catch (Exception e) {
                                        LOG.error(
                                                "UserLabelRichFlatMapFunc Exception: "
                                                        + e.toString());
                                        LOG.error(
                                                "UserLabelRichFlatMapFunc Exception input: "
                                                        + (value.length() >= 100
                                                                ? value.substring(0, 100)
                                                                : value));
                                        e.printStackTrace();
                                    }
                                }
                            });

            out.print();

            env.execute(CaptureUserKeyPropTest.class.toString());
        } catch (Exception e) {
            System.out.println("Exception: " + e.getMessage());
            LOG.error("Exception: " + e.getMessage());
            e.printStackTrace();
        }
    }
}
